package com.SparkCore.RDD.Part

import org.apache.spark.{Partitioner, SparkConf, SparkContext}

/**
 * 分区器
 */
object Spark01_RDD_Part {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(sparkConf)

    val rdd = sc.makeRDD(List(
      ("nba", "xxxxxxxxxxx"),
      ("cba", "xxxxxxxxxxx"),
      ("wnba", "xxxxxxxxxxx"),
      ("nnba", "xxxxxxxxxxx"),
      ("nba", "xxxxxxxxxxx")
    ), 3)

    val partRDD = rdd.partitionBy(new MyPartitioner)
    partRDD.saveAsTextFile("output")

    sc.stop()
  }

  /**
   * 自定义分区器
   * 决定数据到底去哪个分区
   * 1、继承
   */
  class MyPartitioner extends Partitioner{
    //分区的数量
    override def numPartitions: Int = 3
    //返回数的分区索引
    override def getPartition(key: Any): Int = {

      //模式匹配修改版
     /* key match {
        case "nba" => 1
        case "cba" => 2
        case _ => 2
      }*/

      if (key == "nba"){
        0
      } else if (key == "cba"){
        1
      } else {
        2
      }
    }
  }

}
